package com.lenovo.lcdm.integration.config;

import com.lenovo.lcdm.integration.entity.EndpointConfig;
import com.lenovo.lcdm.integration.service.EndpointConfigService;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import java.util.List;

@Configuration
public class KafkaAutoconfiguration implements InitializingBean {

    private final EndpointConfigService endpointConfigService;

    @Autowired
    public KafkaAutoconfiguration(EndpointConfigService endpointConfigService) {
        this.endpointConfigService = endpointConfigService;
    }

    @Override
    public void afterPropertiesSet() {
        List<EndpointConfig> endpointConfigs = endpointConfigService.lambdaQuery().eq(EndpointConfig::getStatus, "active").list();

        List<EndpointConfig> consumers = endpointConfigs.stream().filter(endpointConfig -> "Consumer".equals(endpointConfig.getIntegrationType())).toList();

        consumers.forEach(KafkaConsumerContext::setEndpointConfigMap);
        KafkaConsumerContext.startAllCosumerTask();

        List<EndpointConfig> producers = endpointConfigs.stream().filter(endpointConfig -> "Producer".equals(endpointConfig.getIntegrationType())).toList();
        producers.forEach(KafkaProducerContext::setEndpointConfigMap);

    }

}
